package com.db.transport.server;

import java.io.IOException;
import java.net.ServerSocket;
import java.net.Socket;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.logging.Logger;

import com.db.transport.client.SessionClient;
import com.db.transport.server.listener.SessionServerListener;

public class SessionServer implements Runnable {
	
	private static Logger log = Logger.getLogger(SessionServer.class.getName());
	
	private Set<SessionServerListener> listeners;
	
	private Map<String, SessionClient> clients;
	
	private BlockingQueue<Socket> notAuthorizedSockets;
	
	private ServerSocket providerSocket;

	private Thread threadForNewConnection = new Thread(new Runnable() {

		@Override
		public void run() {
			try {
				while (!Thread.interrupted()) {
					Socket socket = providerSocket.accept();
					log.info("New not authorized client. accept");
		
					try {
						notAuthorizedSockets.put(socket);
					} catch (InterruptedException exception) {
						break;
					}
				}			
			} catch (IOException exception) {
				log.warning("error accept new socket");
			}
		}});
	
	private Thread threadForNewClient = new Thread(new Runnable() {

		@Override
		public void run() {
			while (!Thread.interrupted()) {
				try {
					Socket socket = notAuthorizedSockets.take();
					newNotAuthorizedClient(socket);
				} catch (InterruptedException exception) {
					break;
				}				
			}			
		}});
	
	public SessionServer(int port) throws IOException {
		providerSocket = new ServerSocket(port);
		
		listeners = new HashSet<SessionServerListener>();
		clients = new HashMap<String, SessionClient>();
		notAuthorizedSockets = new LinkedBlockingQueue<Socket>();
	}

	@Override
	public void run() {
		threadForNewConnection.start();
		threadForNewClient.start();
		
		while (!Thread.interrupted()) {
			synchronized (clients) {
				Iterator<SessionClient> iterator = clients.values().iterator();
				while (iterator.hasNext()) {
					SessionClient client = iterator.next();
					String message = client.getMessage();
					if (message != null)
						newMessage(client.getID(), message);
				}
			}
		}
	}

	public void sendMessage(String clientID, String message) {
		log.info("Send message [" + message + "] to [" + clientID + "]");
		synchronized(clients) {
			clients.get(clientID).sendMessage(message);
		}
	}

	private void newMessage(String clientID, String message) {
		log.info("New message [" + message + "] from [" + clientID + "]");
		synchronized(listeners) {
			for (SessionServerListener listener : listeners) {
				listener.newMessage(clientID, message);
			}
		}
	}

	private void newNotAuthorizedClient(Socket socket) {
		log.info("New socket. Try logon.");
		synchronized(listeners) {
			for (SessionServerListener listener : listeners) {
				listener.newNotAuthorizedClient(socket);
			}
		}
	}
	
	public void addListener(SessionServerListener listener) {
		if (listener == null) return;
		
		synchronized(listeners) {
			listeners.add(listener);
		}
	}

	public boolean hasClient(String clientID) {
		synchronized (clients) {
			return clients.containsKey(clientID);
		}
	}

	public void addClient(String clientID, SessionClient client) {
		log.info("Add new client [" + clientID + "]");
		synchronized(clients) {
			clients.put(clientID, client);
		}
	}

}
